package core

import (
	"encoding/json"
	"net"
	"sync"
	"time"

	"gitee.com/huanghua_2017/blade/comm"
	"gitee.com/huanghua_2017/blade/types"
	"github.com/gorilla/websocket"
)

func (slf *Core) readFristRequest(conn *websocket.Conn) {
	conn.SetReadDeadline(time.Now().Add(time.Second * 5))
	_, msg, err := conn.ReadMessage()
	if err != nil || len(msg) <= 2 {
		conn.Close()
		slf.Logger.Printf("读取客户端数据失败，关闭连接 : %+v\n", err)
		return
	}
	msg = comm.DecodeMessage(msg)

	h := types.HelloReqStruct{}
	err = json.Unmarshal(msg, &h)
	if err != nil {
		conn.Close()
		return
	}

	if h.Proto == types.PROXY {
		err = slf.connectUpstream(conn, h.Address)
	} else if h.Proto == types.SLAVE {
		err = slf.startSlave(conn, h.Address)
	} else {
		conn.Close()
		return
	}
	if err != nil {
		res := types.HelloResStruct{Code: -1, ErrMsg: err.Error()}
		wmsg, _ := json.Marshal(res)
		comm.WriteWSMessage(conn, wmsg)
		conn.Close()
	}
}

func (slf *Core) connectUpstream(wsconn *websocket.Conn, addr string) error {
	conn, err := net.Dial("tcp", addr)
	slf.Logger.Printf("连接上游服务器 %s： %+v\n", addr, err)
	if err != nil {
		return err
	}
	res := types.HelloResStruct{}
	wmsg, _ := json.Marshal(res)
	comm.WriteWSMessage(wsconn, wmsg)

	slf.dataExchange(addr, wsconn, conn)
	return nil
}

func (slf *Core) dataExchange(addr string, wconn *websocket.Conn, tconn net.Conn) {
	wsaddr := wconn.RemoteAddr().String()
	tcpaddr := tconn.RemoteAddr().String()
	var once sync.Once
	closed := make(chan struct{}, 1)
	wschan := make(chan []byte, 1000)
	tcpchan := make(chan []byte, 1000)
	slf.State.UpdateALiveConn(1)
	slf.State.UpdateConnsInfo(addr, 1)
	exitfun := func() {
		close(closed)
		slf.State.UpdateALiveConn(-1)
		slf.State.UpdateConnsInfo(addr, -1)
	}

	go func() {
		for {
			select {
			case <-closed:
				wconn.Close()
				slf.Logger.Printf("管道被关闭,关闭下游连接: %s\n", wsaddr)
				return
			default:
				wconn.SetReadDeadline(time.Now().Add(time.Second * 5))
				_, msg, err := wconn.ReadMessage()
				if err != nil {
					slf.Logger.Printf("下游服务器断开连接: %s, %+v\n", wsaddr, err)
					once.Do(exitfun)
					return
				}
				l := len(msg)
				if l > 0 {
					msg = comm.DecodeMessage(msg)
					if msg[0] == '0' {
						wschan <- msg
					} else {
						tcpchan <- msg[1:]
					}
				}
			}
		}
	}()

	go func() {
		for {
			select {
			case <-closed:
				tconn.Close()
				slf.Logger.Printf("管道被关闭,关闭上游连接: %s\n", tcpaddr)
				return
			default:
				data := make([]byte, 4096)
				data[0] = '1'
				tconn.SetReadDeadline(time.Now().Add(time.Second * 1))
				n, err := tconn.Read(data[1:])
				if err != nil && !comm.TimeOutError(err) {
					slf.Logger.Printf("上游服务器断开连接: %s, %+v\n", tcpaddr, err)
					once.Do(exitfun)
					return
				}
				if n > 0 {
					wschan <- data[0 : n+1]
				}
			}
		}
	}()

	go func() {
		for {
			select {
			case <-closed:
				slf.Logger.Printf("%s -> %s 连接已断开, 退出数据发送协程", wsaddr, tcpaddr)
				return
			case wmsg := <-wschan:
				err := comm.WriteWSMessage(wconn, wmsg)
				if err != nil {
					slf.Logger.Printf("发送数据失败: %s, %+v\n", wconn.RemoteAddr().String(), err)
					once.Do(exitfun)
					return
				}
			case tcpmsg := <-tcpchan:
				err := comm.WriteAll(tconn, tcpmsg)
				if err != nil {
					slf.Logger.Printf("发送数据失败: %s, %+v\n", wconn.RemoteAddr().String(), err)
					once.Do(exitfun)
					return
				}
			}
		}
	}()
}
